package com.sk.ccflower.server.common.procManager;

import java.util.Date;
import java.util.Iterator;
import java.util.Timer;
import java.util.concurrent.ConcurrentHashMap;

import javax.jms.MessageConsumer;
import javax.jms.ObjectMessage;

import org.apache.log4j.Logger;

import com.sk.ccflower.lib.commonUtils.DateUtil;
import com.sk.ccflower.lib.define.BaseCfg;
import com.sk.ccflower.lib.define.Consts;
import com.sk.ccflower.server.lib.sysBeans.CpuStatBean;
import com.sk.ccflower.server.lib.sysBeans.ProcInfoBean;
import com.sk.ccflower.server.lib.sysBeans.ProcStatBean;
import com.sk.ccflower.server.lib.sysUtils.SysResCollect;
import com.sun.messaging.ConnectionConfiguration;
import com.sun.messaging.ConnectionFactory;
import com.sun.messaging.Destination;
import com.sun.messaging.jms.Connection;
import com.sun.messaging.jms.JMSException;
import com.sun.messaging.jms.Message;
import com.sun.messaging.jms.Session;


/**
 * @author eunwoogi
 * @since 2009. 11. 11.
 * 프로세스별 정보를 주기적으로 수집해서 전송하는 클래스
 */
public class ProcObserver {
	
	private Logger log;
	CpuStatBean m_cpuStatPrev = null;
	ConcurrentHashMap<String, ProcStatBean> m_hashProcStatPrev = null;
	ConcurrentHashMap<String, ProcInfoBean> m_hashProcInfo = null;
	public ProcInfoScheduler m_sch = null;
	
	public ProcObserver() {
		this.log = Logger.getLogger( getClass() );
		m_hashProcStatPrev = new ConcurrentHashMap<String, ProcStatBean>();
		m_hashProcInfo = new ConcurrentHashMap<String, ProcInfoBean>();
	}
	
	public void run() {
		try {
			// ProcLauncher가 MessageQueue를 통해 전송하는 프로세스 구동 정보를 받기 위한 스레드 구동
			ProcObserverQueueThread thrd = new ProcObserverQueueThread( this, this.m_hashProcInfo );
			thrd.start();
			
			Thread.sleep( 10*1000 );
			/**
			 *  초기 성능 측정을 위해 이전값 생성
			 */
			SysResCollect coll = new SysResCollect();
			if ( null == coll ) {
				log.error( "SysResCollect() is NULL!!" );
			}
			// 전체 CPU 사용율
			this.m_cpuStatPrev = coll.parseCpuStat( coll.getCpuStat() );
			// 프로세스별 PID, CPU/MEM 사용율
			Iterator<String> it = this.m_hashProcInfo.keySet().iterator();
			String key;
			while ( it.hasNext() ) {
				key = it.next();
				ProcInfoBean bean = (ProcInfoBean) this.m_hashProcInfo.get(key);
				bean.pid = coll.getPid( bean.name );
				ProcStatBean statPrev = coll.parseProcStat( coll.getProcStat(bean.pid) );
				
				log.debug( "ProcObsesrver.run() : name("+bean.name+") pid("+bean.pid+")" );
				
				this.m_hashProcStatPrev.putIfAbsent( bean.name, statPrev );
			}
			
			this.m_sch = new ProcInfoScheduler( this );
			
			Timer	timer = new Timer();
			String	ymdhms = DateUtil.getCurrentTime();
			long	fixSt = DateUtil.getUnixTime( Consts.DATE_FT_DT , ymdhms );
			int		collPeriod = BaseCfg.PROC_STAT_PERIOD;
			
			if ( collPeriod <= 0 )	collPeriod = 10;
			
			timer.scheduleAtFixedRate( this.m_sch, new Date(fixSt*1000), (collPeriod*1000) );
		} catch ( Exception e ) {
			e.printStackTrace();
			log.error( e );
		}
	}
	
}

class ProcObserverQueueThread extends Thread {
	
	private Logger log;
	private ProcObserver m_parent = null;
	private ConcurrentHashMap<String, ProcInfoBean> m_hashProcInfo = null;
	private ConnectionFactory m_fac = null;
	private Connection m_connection = null;
	private Session m_session = null;
	private Destination m_dest = null;
	private MessageConsumer m_consumer = null;
	
	public ProcObserverQueueThread( ProcObserver _parent, ConcurrentHashMap<String, ProcInfoBean> hashProcInfo ) {
		this.log = Logger.getLogger( getClass() );
		this.m_parent = _parent;
		this.m_hashProcInfo = hashProcInfo;
	}
	
	private void connectBroker() {
		try {
			m_fac = new ConnectionFactory();
			m_fac.setProperty( ConnectionConfiguration.imqAddressList, "localhost:7676, broker2:6000, broker3:9999" );
			m_fac.setProperty( ConnectionConfiguration.imqReconnectEnabled, "true" );
			
			m_connection = (Connection) m_fac.createConnection();
			
			m_session = (Session) m_connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
			
			m_dest = new com.sun.messaging.Queue( BaseCfg.IPC_MQ_NAME+"_observer" );
			
			m_consumer = m_session.createConsumer( m_dest );
			
			m_connection.start();
		} catch ( Exception e ) {
			try {
				Thread.sleep( 1000 );
			} catch ( Exception e1 ) {
				
			}
			connectBroker();
		}
	}
	
	public void run() {
		ProcInfoBean bean = null;
		
		connectBroker();
		
		while ( true ) {
			try {
				bean = null;
				
				Message inMsg = (Message) m_consumer.receive();
				ObjectMessage objMsg = (ObjectMessage) inMsg;
				
				bean = (ProcInfoBean) objMsg.getObject();
				
				if ( null != bean ) {
					if ( this.m_hashProcInfo.containsKey(bean.name) ) {
						this.m_hashProcInfo.remove(bean.name);
					}
					this.m_hashProcInfo.putIfAbsent(bean.name, bean);
					
					if ( null != this.m_parent ) {
						if ( null != this.m_parent.m_sch ) {
							
						}
					}
					
					log.debug( "ProcObserverQueueThread : SUCCESS receive() -> "+bean.name );
				}
				
				Thread.sleep( 10 );
			} catch ( JMSException e ) {
				try {
					if  ( null != m_session )		m_session.close();
					if  ( null != m_connection )	m_connection.close();
					
					connectBroker();
				} catch ( Exception e1 ) {
					
				}
			} catch ( Exception e ) {
				
			}
		}
	}
	
}
